package com.bigdata.warehouse.dwd;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: com.bigdata.warehouse.dwd
 * @Author: zhangleqing
 * @CreateTime: 2025-08-20  21:58
 * @Description: TODO
 * @Version: 1.0
 */
public class DwdCarsLog {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1.获取Stream的执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 设置检查点
        env.enableCheckpointing(50000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);


        // 修正后的第一个表：ods_entrance_guard_log
        tEnv.executeSql("CREATE TABLE ods_entrance_guard_log (\n" +
                " id STRING,\n" +
                " openMode INT,\n" +
                " openTime STRING,\n" +
                " ownerId INT ,\n" +
                " PRIMARY KEY (id) NOT ENFORCED  // 修正引号错误，移除多余的注释符号\n" +
                ") WITH (\n" +
                " 'connector' = 'mysql-cdc',\n" +
                " 'hostname' = 'node101',\n" +
                " 'port' = '3306',\n" +
                " 'username' = 'root',\n" +
                " 'password' = '123456',\n" +
                " 'database-name' = 'sca',\n" +
                " 'table-name' = 'entrance_guard_log'\n" +
                ")\n");
        //System.out.println("第一个的数据：");
        //tEnv.executeSql("SELECT * FROM ods_entrance_guard_log limit 10").print();


        // 修正后的第二个表：ods_cars_log
        tEnv.executeSql("CREATE TABLE ods_cars_log (\n" +
                " id STRING,\n" +
                " opTime STRING,\n" +
                " ctype INT,\n" +
                " carCode STRING,\n" +
                " cId INT,\n" +
                " PRIMARY KEY (id) NOT ENFORCED  // 修正引号错误，移除多余的注释符号\n" +
                ") WITH (\n" +
                " 'connector' = 'mysql-cdc',\n" +
                " 'hostname' = 'node101',\n" +
                " 'port' = '3306',\n" +
                " 'username' = 'root',\n" +
                " 'password' = '123456',\n" +
                " 'database-name' = 'sca',\n" +
                " 'table-name' = 'cars_Log'\n" +
                ")\n");
        //System.out.println("第二个的数据：");
        //tEnv.executeSql("SELECT * FROM ods_cars_log limit 10").print();



        tEnv.executeSql("CREATE TABLE ods_community (\n" +
                "    id int,\n" +
                "    communityName string,\n" +
                "    district_id int,\n" +
                "    city_id int,\n" +
                "    communityLocation string,\n" +
                "    remark string,\n" +
                "    PRIMARY KEY (id) NOT ENFORCED\n" +
                ")WITH (\n" +
                "    'connector' = 'mysql-cdc',\n" +
                "    'hostname' = 'node101',\n" +
                "    'port' = '3306',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'database-name' = 'sca_doris',\n" +
                "    'table-name' = 'community'\n" +
                ")");

        tEnv.executeSql("select * from ods_community limit 10").print();



        //env.execute("DwdCarsLog");
    }
}
